import json
import random
import time
from datetime import datetime, timedelta
from kafka import KafkaProducer

# Kafka配置
producer = KafkaProducer(
    bootstrap_servers='node1:9092',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

# 受害者IP范围
victim_ips = [f"192.168.1.{i}" for i in range(10, 50)]
# 攻击类型
attack_types = ["TCP_SYN_FLOOD", "UDP_FLOOD", "HTTP_FLOOD", "DNS_AMP", "NTP_AMP"]
# 风险等级
risk_levels = ["LOW", "MEDIUM", "HIGH", "CRITICAL"]
# 流量方向
directions = ["inbound", "outbound"]


def generate_attack_data():
    """生成模拟攻击数据"""
    event_time = datetime.now() - timedelta(minutes=random.randint(0, 60))

    return {
        "victim_ip": random.choice(victim_ips),
        "device_ip": f"10.0.0.{random.randint(1, 5)}",
        "ddos_total_kbps": random.uniform(100, 10000),
        "http_pps": random.uniform(50, 5000),
        "tcp_pps": random.uniform(30, 3000),
        "udp_pps": random.uniform(20, 2000),
        "net_direction": random.choice(directions),
        "time__": event_time.isoformat(timespec='milliseconds'),
        "event_type_l2": random.choice(attack_types),
        "risk_level": random.choice(risk_levels),
        "is_critical": random.random() > 0.8  # 20%概率为严重攻击
    }


# 发送数据到Kafka
for i in range(1000):
    data = generate_attack_data()
    producer.send('attack_monitor_topic', value=data)

    # 打印前10条数据示例
    if i < 10:
        print(f"Sample data {i + 1}: {json.dumps(data, indent=2)}")

    time.sleep(0.1)  # 控制发送速率

producer.flush()
print(f"Successfully sent 1000 attack records to Kafka topic 'attack_monitor_topic'")